package com.baishancloud.log.live.sanma

import com.baishancloud.log.format.scala.SanmaOrigin
import org.apache.flink.api.common.functions.RichFilterFunction
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.configuration.Configuration

import java.util.{Objects, Timer, TimerTask}

/**
 * 过滤数据
 *
 * @author ziqiang.wang 
 * @date 2022/3/31 16:04
 */
class DataFilter(parameterTool: ParameterTool) extends RichFilterFunction[SanmaOrigin] {

  var allowOldest: Long = System.currentTimeMillis() / 1000 - 3600 * parameterTool.getLong(allowOldestHour, 6)

  override def open(parameters: Configuration): Unit = {
    // 1分钟更新一次最大允许迟到时间
    new Timer("allowOldestUpdate", true).scheduleAtFixedRate(new TimerTask {
      override def run(): Unit = allowOldest = System.currentTimeMillis() / 1000 - 3600 * parameterTool.getLong(allowOldestHour, 6)
    }, 60000, 60000)
  }

  override def filter(value: SanmaOrigin): Boolean = {
    val logTime: Long = value.datetimeOfUTC().toEpochSecond
    //日志时间在6小时以内
    val b1: Boolean = logTime > allowOldest
    //${connect-status}值为1（即正在拉流）
    val b2: Boolean = Objects.equals(value.connectStatus(), "1")
    //${x-comment}值为-，取边缘
    val b3 = Objects.equals(value.xComment(), "-")
    b1 && b2 && b3
  }
}
